package com.iot.processor.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 数据生产者
 * 模拟IoT设备生成符合协议的十六进制消息
 */
public class DataProducer implements Runnable {
    
    private static final Logger logger = LoggerFactory.getLogger(DataProducer.class);
    
    private final BlockingQueue<String> messageQueue;
    private final AtomicBoolean running;
    private final AtomicLong messageCount;
    
    // 生产者配置
    private final int deviceIdStart;
    private final int deviceIdEnd;
    private final int messagesPerSecond;
    private final String producerName;
    
    // 温湿度范围配置
    private static final double MIN_TEMPERATURE = -10.0;  // 最低温度
    private static final double MAX_TEMPERATURE = 45.0;   // 最高温度
    private static final double MIN_HUMIDITY = 20.0;      // 最低湿度
    private static final double MAX_HUMIDITY = 95.0;      // 最高湿度
    
    /**
     * 构造函数
     * 
     * @param messageQueue 消息队列
     * @param producerName 生产者名称
     * @param deviceIdStart 设备ID起始值
     * @param deviceIdEnd 设备ID结束值
     * @param messagesPerSecond 每秒生成消息数量
     */
    public DataProducer(BlockingQueue<String> messageQueue, String producerName, 
                       int deviceIdStart, int deviceIdEnd, int messagesPerSecond) {
        this.messageQueue = messageQueue;
        this.producerName = producerName;
        this.deviceIdStart = deviceIdStart;
        this.deviceIdEnd = deviceIdEnd;
        this.messagesPerSecond = messagesPerSecond;
        this.running = new AtomicBoolean(false);
        this.messageCount = new AtomicLong(0);
    }
    
    @Override
    public void run() {
        running.set(true);
        logger.info("数据生产者 {} 开始运行，设备ID范围: {}-{}, 每秒生成消息: {}", 
                   producerName, deviceIdStart, deviceIdEnd, messagesPerSecond);
        
        long intervalNanos = 1_000_000_000L / messagesPerSecond;  // 每条消息的间隔时间（纳秒）
        long lastTime = System.nanoTime();
        
        while (running.get()) {
            try {
                long currentTime = System.nanoTime();
                
                // 控制生产速率
                if (currentTime - lastTime >= intervalNanos) {
                    String message = generateMessage();
                    
                    // 非阻塞方式放入队列，如果队列满了就跳过
                    if (messageQueue.offer(message)) {
                        messageCount.incrementAndGet();
                        lastTime = currentTime;
                    } else {
                        logger.warn("消息队列已满，跳过消息: {}", message);
                        Thread.sleep(1); // 短暂休眠避免CPU占用过高
                    }
                } else {
                    // 精确控制时间间隔
                    long sleepNanos = intervalNanos - (currentTime - lastTime);
                    if (sleepNanos > 1000000) { // 大于1毫秒才休眠
                        Thread.sleep(sleepNanos / 1000000, (int)(sleepNanos % 1000000));
                    }
                }
                
            } catch (InterruptedException e) {
                logger.info("数据生产者 {} 被中断", producerName);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                logger.error("数据生产者 {} 发生异常", producerName, e);
            }
        }
        
        logger.info("数据生产者 {} 停止运行，总共生成消息: {}", producerName, messageCount.get());
    }
    
    /**
     * 生成符合协议的十六进制消息
     * 格式：设备ID(8位) + 时间戳(8位) + 温度(4位) + 湿度(4位)
     * 
     * @return 十六进制消息字符串
     */
    private String generateMessage() {
        Random random = ThreadLocalRandom.current();
        
        // 生成设备ID（在指定范围内）
        int deviceId = random.nextInt(deviceIdEnd - deviceIdStart + 1) + deviceIdStart;
        String deviceIdHex = String.format("%08X", deviceId);
        
        // 生成当前时间戳
        long timestamp = System.currentTimeMillis() / 1000;
        String timestampHex = String.format("%08X", (int)timestamp);
        
        // 生成温度（转换为整数，乘以100）
        double temperature = MIN_TEMPERATURE + (MAX_TEMPERATURE - MIN_TEMPERATURE) * random.nextDouble();
        int temperatureInt = (int)(temperature * 100);
        String temperatureHex = String.format("%04X", temperatureInt);
        
        // 生成湿度（转换为整数，乘以100）
        double humidity = MIN_HUMIDITY + (MAX_HUMIDITY - MIN_HUMIDITY) * random.nextDouble();
        int humidityInt = (int)(humidity * 100);
        String humidityHex = String.format("%04X", humidityInt);
        
        // 组合完整消息
        String message = deviceIdHex + timestampHex + temperatureHex + humidityHex;
        
        logger.debug("生成消息: {} (设备ID: {}, 温度: {:.2f}°C, 湿度: {:.2f}%)", 
                    message, deviceId, temperature, humidity);
        
        return message;
    }
    
    /**
     * 停止生产者
     */
    public void stop() {
        running.set(false);
        logger.info("数据生产者 {} 收到停止信号", producerName);
    }
    
    /**
     * 检查生产者是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }
    
    /**
     * 获取已生成的消息数量
     */
    public long getMessageCount() {
        return messageCount.get();
    }
    
    /**
     * 获取生产者名称
     */
    public String getProducerName() {
        return producerName;
    }
    
    /**
     * 获取生产者状态信息
     */
    public String getStatus() {
        return String.format("生产者 %s - 运行状态: %s, 已生成消息: %d, 设备范围: %d-%d", 
                           producerName, running.get() ? "运行中" : "已停止", 
                           messageCount.get(), deviceIdStart, deviceIdEnd);
    }
}